package org.nbict.iot.trident.kafka;


import org.apache.storm.kafka.trident.selector.KafkaTopicSelector;
import org.apache.storm.trident.tuple.TridentTuple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Created by songseven on 18/7/6.
 */
public class FieldNameTopicSelector implements KafkaTopicSelector {


    private static final Logger LOG = LoggerFactory.getLogger(org.apache.storm.kafka.bolt.selector.FieldNameTopicSelector.class);
    private final String fieldName;
    private final String defaultTopicName;

    public FieldNameTopicSelector(String fieldName, String defaultTopicName) {
        this.fieldName = fieldName;
        this.defaultTopicName = defaultTopicName;
    }

    public String getTopic(TridentTuple tuple) {
        if(tuple.contains(this.fieldName)) {
            return tuple.getStringByField(this.fieldName);
        } else {
            LOG.warn("Field " + this.fieldName + " Not Found . Using default topic " + this.defaultTopicName);
            return this.defaultTopicName;
        }
    }
}
